package com.cjz.demo5;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.schedulers.Schedulers;

/**
 * RxJava的基本使用
 */

public class Test {
    public static void main(String[] args) throws InterruptedException {

        System.out.println("main() thread id " + Thread.currentThread().getId());

        testRxJava1();
        testRxJava2();
        testRxJava3();
        testRxJava4();

        System.out.println("main() thread id " + Thread.currentThread().getId());

        Thread.sleep(10_000);
    }

    private static void testRxJava4() {
        Observable
                .create((Subscriber<? super Integer> subscriber) -> {
                    System.out.println("OnSubscribe call() thread id " + Thread.currentThread().getId());
                    for (int i = 0; i < 10; i++) {
                        subscriber.onNext(i);
                    }
                    subscriber.onCompleted();
                })
                .subscribeOn(Schedulers.newThread())
                .subscribe(
                        integer -> System.out.println("Subscriber call() thread id " + Thread.currentThread().getId() + " current download progress " + integer),//
                        error -> System.out.println("error " + error.getMessage()),//
                        () -> System.out.println("complete")//
                )
        ;
    }

    private static void testRxJava3() {
        Observable
                .create((Subscriber<? super Integer> subscriber) -> {
                    System.out.println("OnSubscribe call() thread id " + Thread.currentThread().getId());
                    for (int i = 0; i < 10; i++) {
                        subscriber.onNext(i);
                    }
                    subscriber.onCompleted();
                })
                .subscribeOn(Schedulers.newThread())
                .subscribe(integer -> {
                    System.out.println("Subscriber call() thread id " + Thread.currentThread().getId() + " current download progress " + integer);
                })
        ;
    }

    private static void testRxJava2() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {

                        System.out.println("OnSubscribe call() thread id " + Thread.currentThread().getId());
                        for (int i = 0; i < 10; i++) {
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("Subscriber call() thread id " + Thread.currentThread().getId() + " downloaded !");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("Subscriber call() thread id " + Thread.currentThread().getId() + " current download progress " + integer);
                    }
                });
    }

    private static void testRxJava1() {
        //创建被观察对象
        Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {

                System.out.println("OnSubscribe call() thread id " + Thread.currentThread().getId());
                for (int i = 0; i < 10; i++) {
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        });

        //创建订阅者
        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

                System.out.println("Subscriber call() thread id " + Thread.currentThread().getId() + " downloaded !");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("Subscriber call() thread id " + Thread.currentThread().getId() + " current download progress " + integer);
            }
        };

        observable = observable.subscribeOn(Schedulers.newThread());

        //桥接被观察对象与订阅者
        observable.subscribe(subscriber);
    }
}
